package com.icow.web;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;

import javax.servlet.http.HttpServletRequest;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.messaging.Message;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.icow.dao.JiandaoyunDao;
import com.icow.domain.JiandaoyunReq;
import com.icow.kafka.KafkaIcowMsg;

import lombok.extern.slf4j.Slf4j;

@RestController
@Slf4j
public class JiandaoyunController {
	@Autowired
	private JiandaoyunDao jiandaoyunDao;

	private static Map<Integer, Integer> pointMap;
	private static Map<Integer, Integer> lineMap;

	// "拉号" key
	@Value("${icow.key-lahao}")
	private String keyLahao;
	// "完成数" key
//	@Value("${icow.key-num}")
//	private String keyNum;

	@Autowired
	private KafkaTemplate<String, KafkaIcowMsg> kafka;

	static {
		pointMap = new HashMap<>();
		pointMap.put(279, 2002);
		pointMap.put(283, 2002);
		pointMap.put(286, 2002);
		pointMap.put(297, 2010);
		pointMap.put(202, 2010);
		pointMap.put(205, 2010);
		pointMap.put(204, 2010);
		pointMap.put(207, 2010);
		pointMap.put(201, 2010);
		pointMap.put(208, 2010);
		pointMap.put(291, 2012);
		pointMap.put(292, 2012);
		pointMap.put(295, 2012);
		pointMap.put(294, 2012);
		pointMap.put(285, 2012);
		pointMap.put(281, 2012);
		pointMap.put(282, 2012);
		pointMap.put(284, 2012);
		pointMap.put(278, 2013);
		pointMap.put(274, 2013);
		pointMap.put(277, 2013);
		pointMap.put(276, 1016);
		pointMap.put(273, 1016);
		pointMap.put(269, 1016);
		pointMap.put(256, 1032);
		pointMap.put(253, 1032);
		pointMap.put(263, 1032);
		pointMap.put(255, 1037);
		pointMap.put(252, 1037);
		pointMap.put(261, 1037);
		pointMap.put(265, 1037);
		pointMap.put(264, 1038);
		pointMap.put(267, 1038);
		pointMap.put(268, 1038);
		pointMap.put(271, 1038);
		pointMap.put(272, 1038);
		pointMap.put(275, 1038);
		pointMap.put(233, 1022);
		pointMap.put(236, 1022);
		pointMap.put(239, 1022);
		pointMap.put(243, 1022);
		pointMap.put(246, 1022);
		pointMap.put(249, 1022);
		pointMap.put(231, 1023);
		pointMap.put(234, 1023);
		pointMap.put(237, 1023);
		pointMap.put(235, 1023);
		pointMap.put(238, 1023);
		pointMap.put(241, 1025);
		pointMap.put(244, 1025);
		pointMap.put(247, 1025);
		pointMap.put(242, 1025);
		pointMap.put(245, 1025);
		pointMap.put(248, 1025);

		lineMap = new HashMap<>();
		lineMap.put(2002, 1);
		lineMap.put(2010, 1);
		lineMap.put(2012, 1);
		lineMap.put(2013, 1);
		lineMap.put(1016, 2);
		lineMap.put(1032, 2);
		lineMap.put(1037, 2);
		lineMap.put(1038, 2);
		lineMap.put(1022, 3);
		lineMap.put(1023, 3);
		lineMap.put(1025, 3);
	}

	@RequestMapping("/push")
	public String receive(@RequestBody(required = false) String req, HttpServletRequest request) {
		log.info("====== method: {}, receive data: {}", request.getMethod(), req);
		if (req == null) {
			return "failed, no data";
		}
		JiandaoyunReq domain = JSON.parseObject(req, JiandaoyunReq.class);
		String op = domain.getOp();
		if ("data_create".equals(op)) {
			JSONObject dataJson = domain.getData();
			Integer lahao = dataJson.getInteger(keyLahao);
			Integer point = pointMap.get(lahao);
			Integer line = lineMap.get(point);
			KafkaIcowMsg msg = new KafkaIcowMsg();
			msg.setLahao(lahao);
			msg.setPoint(point);
			msg.setLine(line);
			msg.setNum(1); // 数量用不到了
			log.info("====== send: {}", msg);
			kafka.sendDefault(msg);
		}
		jiandaoyunDao.save(domain);

		return "success";
	}

	@RequestMapping("/test")
	public String testPush(KafkaIcowMsg msg) {
		Integer lahao = msg.getLahao();
		Integer point = pointMap.get(lahao);
		if (point == null) {
			return "fail!";
		}
		Integer line = lineMap.get(point);
		msg.setPoint(point);
		msg.setLine(line);
		msg.setNum(1);
		log.info("======send: {}", msg);
		kafka.sendDefault(msg);
		return "success";
	}

	@RequestMapping("/clear")
	public String clear() {
		jiandaoyunDao.deleteAll();
		return "success";
	}

	@RequestMapping("/all")
	public List<JiandaoyunReq> listAll() {
		List<JiandaoyunReq> list = jiandaoyunDao.findAll();
		return list;
	}

	@Autowired
	private KafkaTemplate<String, String> kafkaTemplate;

	@RequestMapping("/t1")
	public String t1() throws Exception {
		ListenableFuture<SendResult<String, String>> f = kafkaTemplate.send("test", "" + System.currentTimeMillis());
		SendResult<String, String> sr = f.get();
		String s = sr.toString();

		return "success:" +s;
	}

	/**
	 * 1号线，2010点
	 */
	@RequestMapping("/t2")
	public String t2() {
		KafkaIcowMsg msg = new KafkaIcowMsg();
		msg.setLahao(297);
		msg.setNum(2);

		Integer lahao = msg.getLahao();
		Integer point = pointMap.get(lahao);
		Integer line = lineMap.get(point);
		msg.setPoint(point);
		msg.setLine(line);
		kafka.sendDefault(msg);
		return "success";
	}

	/**
	 * 1号线，2012点
	 */
	@RequestMapping("/t3")
	public String t3() {
		KafkaIcowMsg msg = new KafkaIcowMsg();
		msg.setLahao(291);
		msg.setNum(1);

		Integer lahao = msg.getLahao();
		Integer point = pointMap.get(lahao);
		Integer line = lineMap.get(point);
		msg.setPoint(point);
		msg.setLine(line);
		kafka.sendDefault(msg);
		return "success";
	}

	@KafkaListener(topics = "test")
	public void processMessage(Message<String> content) {
		log.info("======receive: {}", content);
	}

	@KafkaListener(topics = "${spring.kafka.template.default-topic}")
	public void processMessage(KafkaIcowMsg msg) {
		log.info("======receive: {}", msg);
	}
}
